package com.navinfo.platform.dataanalysis.service.impl;

import com.mongodb.spark.rdd.api.java.JavaMongoRDD;
import com.navinfo.platform.common.dto.DataInfo;
import com.navinfo.platform.common.service.IDataHandleService;
import com.navinfo.platform.dataanalysis.dto.IdleBean;
import com.navinfo.platform.dataanalysis.service.ILocationDataBussinessService;
import com.navinfo.platform.dataanalysis.service.IVehicleBaseStatisticsDayService;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.storage.StorageLevel;
import org.bson.Document;
import scala.Tuple2;

import java.util.List;
import java.util.Map;

public enum DataAnalysisHandleService implements IDataHandleService {
    /**
     *
     */
    INSTANCE;
    private IVehicleBaseStatisticsDayService statisticsDayService = VehicleBaseStatisticsDayService.INSTANCE;
    private ILocationDataBussinessService locationDataBussinessService = LocationDataBusinessService.INSTANCE;

    @SuppressWarnings("unchecked")
    @Override
    public void handle(List<Object> obj, Map<String, String> configMap, SparkSession spark) {
        Dataset<Row> locationDataset = (Dataset<Row>) obj.get(0);
        Dataset<Row> realtimeDataset = (Dataset<Row>) obj.get(1);
        Dataset<Row> tileDataset = (Dataset<Row>) obj.get(2);
//        JavaMongoRDD<Document> javaMongoRDD = (JavaMongoRDD<Document>) obj.get(2);
        //将0200数据转成排序后的<tid,List<LocationDataPoint>>
        JavaPairRDD<Long, List<DataInfo>> locationDataPair = statisticsDayService.convert(locationDataset,
                configMap);
        locationDataPair.persist(StorageLevel.MEMORY_AND_DISK());

        JavaPairRDD<Long, List<DataInfo>> realtimeDataRDD = locationDataBussinessService.convert(realtimeDataset,
                configMap);
        //整合数据
        JavaPairRDD<Long, List<DataInfo>> allDataRDD = locationDataBussinessService.unionData(locationDataPair, realtimeDataRDD);
        allDataRDD.persist(StorageLevel.MEMORY_AND_DISK());

        //怠速数据
        JavaRDD<IdleBean> idleBeanJavaRDD = IdleSpeedService.INSTANCE.calcIdleInfo(allDataRDD);
        String[] fields = {"idleDuration", "idleTimes"};
        Dataset<Row> idleDataset = spark.createDataFrame(idleBeanJavaRDD, IdleBean.class).select("tid", fields);
        idleDataset.persist(StorageLevel.MEMORY_AND_DISK());

        Dataset<Row> vehicleBaseStatisticsDay = statisticsDayService.handle(locationDataPair,idleDataset,tileDataset,configMap,
                spark);

        ////统计事件、告警、故障信息
        Dataset<Row> basicInfo = vehicleBaseStatisticsDay.select("tid","mileage", "workDuration");
        basicInfo.persist(StorageLevel.MEMORY_AND_DISK());
        vehicleBaseStatisticsDay.unpersist();
        locationDataBussinessService.statisticInfo(spark, locationDataPair, allDataRDD, basicInfo, idleDataset, null, configMap);

    }


}
